void init_mwindowjoin()         // initialize structures
{
  memset(windowjoin_nodes, 0, sizeof(windowjoin_nodes));
  memset(windowjoin_windows, 0, sizeof(windowjoin_windows));
  memset(windowjoin_pairs, 0, sizeof(windowjoin_pairs));
}

void display_joinbuffer()               // display window for debugging
{
  int i, j, k, len;
  char s[1024];

  DBGOUT("Nodes:");
  for (i = 0; i < MAX_JOINNODES; i++)
    if (windowjoin_nodes[i].used)
    {
      snprintf(s, sizeof(s), "%d", windowjoin_nodes[i].node_id);
      print_path(windowjoin_nodes[i].path, windowjoin_nodes[i].path_len, s);
    }

  DBGOUT("Windows:");
  DBGOUT(" Id Node Query Rel Seq Tuple");
  for (i = 0; i < MAX_JOINWINDOWS; i++)
    if (windowjoin_windows[i].used)
    {
      snprintf(s, sizeof(s), "%3d  %3d   %3d %3d %3d ", i,
        windowjoin_nodes[windowjoin_windows[i].node_index].node_id,
        windowjoin_windows[i].query_id, windowjoin_windows[i].rel_id, windowjoin_windows[i].max_seq);
        
      if (windowjoin_windows[i].window)
        for (j = 0; j < windowjoin_windows[i].height; j++)
        {
          len = strlen(s);
          sprintf(&s[len], "(");
          for (k = TUPLEID_LEN; k < windowjoin_windows[i].width; k++)
          {
            len = strlen(s);
            if (windowjoin_windows[i].window[j*windowjoin_windows[i].width] == 0xffff)
              sprintf(&s[len], "%5s ", " ");
            else sprintf(&s[len], "%5d ", windowjoin_windows[i].window[j*windowjoin_windows[i].width+k]);
          }
          len = strlen(s);
          sprintf(&s[len], ") ");
        }
        
      DBGOUT(s);
    }

  DBGOUT("Pairs:");
  DBGOUT(" Src  Trg  Opt Ssel Tsel Jsel Stup Ttup Jtup  Time");
  for (i = 0; i < MAX_JOINPAIRS; i++)
    if (windowjoin_pairs[i].used)
      DBGOUT("%4d %4d %4d %4d %4d %4d %4d %4d %4d %5ld", windowjoin_pairs[i].window_index[REL_S],
        windowjoin_pairs[i].window_index[REL_T], windowjoin_pairs[i].options,
        windowjoin_pairs[i].sel_est[REL_S], windowjoin_pairs[i].sel_est[REL_T],
        windowjoin_pairs[i].sel_est[REL_J], windowjoin_pairs[i].tuple_count[REL_S],
        windowjoin_pairs[i].tuple_count[REL_T], windowjoin_pairs[i].tuple_count[REL_J],
        windowjoin_pairs[i].start_time);
}

int get_nodeindex(ADDRESS node_id)                 // search the join buffer by node id
{
  int i;

  for (i = 0; i < MAX_JOINNODES; i++)
    if (windowjoin_nodes[i].used && windowjoin_nodes[i].node_id == node_id)
      return i;

  return -1;
}

int get_freenodeindex()                          // find a free slot in the join buffer
{
  int i;

  for (i = 0; i < MAX_JOINNODES; i++)
    if (windowjoin_nodes[i].used == 0)
      return i;

  return -1;
}

int add_node(ADDRESS node_id, uint8_t path_len, ADDRESS *path)
{
  int ni = get_nodeindex(node_id);

  if (ni < 0)
  {
    ni = get_freenodeindex();
    if (ni < 0)
    {
      DBGERR("Cannot allocate node");
      pause_sim();
      return -1;
    }

    windowjoin_nodes[ni].used = 1;
    windowjoin_nodes[ni].path_len = 0;
    windowjoin_nodes[ni].path = NULL;
    windowjoin_nodes[ni].node_id = node_id;
  }

  if (path_len > 0 && (windowjoin_nodes[ni].path_len == 0 || windowjoin_nodes[ni].path_len > path_len))
  {
    windowjoin_nodes[ni].path_len = path_len;
    windowjoin_nodes[ni].path = (ADDRESS *) mrealloc(windowjoin_nodes[ni].path, path_len*sizeof(ADDRESS));
    memcpy(windowjoin_nodes[ni].path, path, path_len*sizeof(ADDRESS));
  }

  return ni;
}

int get_windowindex(uint8_t node_index, uint8_t query_id, uint8_t rel_id)
{
  int i;

  for (i = 0; i < MAX_JOINWINDOWS; i++)
    if (windowjoin_windows[i].used && windowjoin_windows[i].node_index == node_index &&
      windowjoin_windows[i].query_id == query_id && windowjoin_windows[i].rel_id == rel_id)
      return i;

  return -1;
}

int get_freewindowindex()
{
  int i;

  for (i = 0; i < MAX_JOINWINDOWS; i++)
    if (windowjoin_windows[i].used == 0)
      return i;

  return -1;
}

int alloc_window(uint8_t query_id, uint8_t rel_id, uint8_t node_index)
{
  int wi = get_windowindex(node_index, query_id, rel_id);
  WINDOW_ENTRY *p;

  if (wi < 0)
  {
    wi = get_freewindowindex();
    if (wi < 0)
    {
      DBGERR("Cannot allocate window for node %d", windowjoin_nodes[node_index].node_id);
      pause_sim();
      return -1;
    }
    p = &windowjoin_windows[wi];

    p->used = 1;
    p->query_id = query_id;
    p->rel_id = rel_id;
    p->node_index = node_index;
    p->width = query_buffer[query_id].tuple_len[rel_id];
    p->height = query_buffer[query_id].spec.params[PARAM_WINDOWSIZE];
    p->window = NULL;
    p->max_seq = 0x7f;
  }

  return wi;
}

void remove_window(int window_index)
{
  if (window_index < 0)
    return;

  if (windowjoin_windows[window_index].window)
  {
    mfree(windowjoin_windows[window_index].window);
    windowjoin_windows[window_index].window = NULL;
  }
  windowjoin_windows[window_index].used = 0;
}

int get_joinpair(uint8_t *window_index)
{
  int i, j;
  uint8_t found;

  for (i = 0; i < MAX_JOINPAIRS; i++)
    if (windowjoin_pairs[i].used)
    {
      found = 1;
      for (j = 0; j < SRCREL_COUNT; j++)
        if (windowjoin_pairs[i].window_index[j] != window_index[j])
        {
          found = 0;
          break;
        }

      if (found)
        return i;
    }

  return -1;
}

int get_freejoinpair()
{
  int i;

  for (i = 0; i < MAX_JOINPAIRS; i++)
    if (windowjoin_pairs[i].used == 0)
      return i;

  return -1;
}

int add_joinpair(uint8_t query_id, ADDRESS node1, uint8_t path_len1, ADDRESS *path1, ADDRESS node2,
  uint8_t path_len2, ADDRESS *path2, uint8_t options)
{                                              // register a join pair between a source and a target
  int pi, wi1, wi2, ni1 = add_node(node1, path_len1, path1), ni2 = add_node(node2, path_len2, path2);
  uint8_t index_pair[SRCREL_COUNT];
  JOINPAIRS_ENTRY *p;

  if (query_buffer[query_id].used == 0)
  {
    DBGERR("Query %d not used", query_id);
    return -1;
  }

  if (ni1 < 0 || ni2 < 0)
  {
    DBGERR("Cannot allocate node entries!");
    pause_sim();
    return -1;
  }

  wi1 = alloc_window(query_id, REL_S, ni1);
  wi2 = alloc_window(query_id, REL_T, ni2);
  if (wi1 < 0 || wi2 < 0)
  {
    DBGERR("Cannot allocate join windows!");
    pause_sim();
    return -1;
  }

  index_pair[REL_S] = wi1;
  index_pair[REL_T] = wi2;
  pi = get_joinpair(index_pair);
  if (pi < 0)
  {
    pi = get_freejoinpair();
    if (pi < 0)
    {
      DBGERR("Cannot get a free join pair!");
      pause_sim();
      return -1;
    }

    p = &windowjoin_pairs[pi];
    p->used = 1;
    memcpy(p->window_index, index_pair, sizeof(index_pair));
    p->options = 0;
    memcpy(p->sel_est, query_buffer[query_id].sel_est, sizeof(p->sel_est));
    memset(p->tuple_count, 0, sizeof(p->tuple_count));
    p->start_time = timer_cycles;
  }

  windowjoin_pairs[pi].options |= options;
  return 0;
}

void remove_joinpair(uint8_t query_id, ADDRESS node1, ADDRESS node2, uint8_t options, uint8_t confirm)
{                                             // unregister a pair
  int i, ni1 = get_nodeindex(node1), ni2 = get_nodeindex(node2), wi1, wi2, pi;
  uint8_t index_pair[SRCREL_COUNT];
  uint8_t window_used;

  if (query_buffer[query_id].used == 0)
  {
    DBGERR("Query %d not used", query_id);
    return;
  }

  if (ni1 < 0 || ni2 < 0)
  {
    if (confirm)
    {
      DBGERR("Cannot remove pair! Nodes do not exist");
      pause_sim();
    }
    return;
  }

  wi1 = get_windowindex(ni1, query_id, REL_S);
  wi2 = get_windowindex(ni2, query_id, REL_T);
  if (wi1 < 0 || wi2 < 0)
  {
    if (confirm)
    {
      DBGERR("Cannot remove pair! Windows do not exist");
      pause_sim();
    }
    return;
  }

  index_pair[REL_S] = wi1;
  index_pair[REL_T] = wi2;
  pi = get_joinpair(index_pair);
  if (pi < 0)
  {
    if (confirm)
    {
      DBGERR("Cannot remove pair! Pair does not exist");
      pause_sim();
    }
    return;
  }

  windowjoin_pairs[pi].options &= (~options);
  if (windowjoin_pairs[pi].options == 0)
  {
    DBGOUT("Removing pair (%d, %d)", node1, node2);
    windowjoin_pairs[pi].used = 0;
  }

  for (wi1 = 0; wi1 < MAX_JOINWINDOWS; wi1++)
    if (windowjoin_windows[wi1].used)
    {
      window_used = 0;
      for (pi = 0; i < MAX_JOINPAIRS; pi++)
        if (windowjoin_pairs[pi].used)
        {
          for (i = 0; i < SRCREL_COUNT; i++)
            if (windowjoin_pairs[pi].window_index[i] == wi1)
            {
              window_used = 1;
              break;
            }
          if (window_used)
            break;
        }

      if (window_used == 0)
      {
        DBGOUT("Removing unused window %d", wi1);
        remove_window(wi1);
      }
    }
}

void do_join(QUERY_ENTRY *q, uint8_t window_index, uint8_t tuple_index) // once a value has been placed in the window
{                                                         // try to join with with values from other sources/targets
  int i, j, k, wi;
  uint8_t tuple_len;
  ATTRIBUTE tuple[MAX_TUPLELEN], *p_s, *p_t;
  TUPLE_MSGHI msg;
  WINDOW_ENTRY *p1, *p2;

DO_STACKTEST

  msg.cmd = CMDHI_TUPLE;
  msg.cache_multicast = 0;
  page_init(&msg.page);
  tuple_len = q->tuple_len[REL_J];
  TUPLESET(tuple, TOS_NODE_ID, 0, 0, q->query_id, REL_J, 0);

  p1 = &windowjoin_windows[window_index];
  for (i = 0; i < MAX_JOINPAIRS; i++)
    if (windowjoin_pairs[i].used)
      if (windowjoin_pairs[i].window_index[p1->rel_id] == window_index)
      {
        wi = windowjoin_pairs[i].window_index[1-p1->rel_id];
        p2 = &windowjoin_windows[wi];
        if (p2->window == NULL)
          continue;

        for (j = 0; j < p2->height; j++)
          if (p2->window[j*p2->width] != 0xffff)
          {
            if (p1->rel_id == REL_S)
            {
              p_s = &p1->window[tuple_index*p1->width+TUPLEID_LEN];
              p_t = &p2->window[j*p2->width+TUPLEID_LEN];
            } else {
              p_t = &p1->window[tuple_index*p1->width+TUPLEID_LEN];
              p_s = &p2->window[j*p2->width+TUPLEID_LEN];
            }

            if (eval_stack(&q->spec.rel[REL_DYNAMIC][REL_J].pred, p_s, p_t, 0))
            {
              DBGOUT("Tuple %d window %d joins with tuple %d window %d", tuple_index, window_index, j, wi);
              if (windowjoin_pairs[i].tuple_count[REL_J] < MAX_ESTTUPLECOUNT)
                windowjoin_pairs[i].tuple_count[REL_J]++;
              for (k = TUPLEID_LEN; k < tuple_len; k++)
                tuple[k] = eval_stack(&q->spec.rel[REL_DYNAMIC][REL_J].eval[k-TUPLEID_LEN].eval, p_s, p_t, 0);

              addtuple_wrap(&msg.page, tuple_len, tuple);
              send_tuplemsg(&msg);
              page_init(&msg.page);

#ifdef TOSSIM
              log_queryjoinout(BASESTATION_ADDR, tuple_len, tuple);
#endif
            }
          }
      }
}

uint8_t place_tuple(uint8_t tuple_len, ATTRIBUTE *tuple)
{                    // place incoming tuple into the window join, if disable_join == 0, do not produce output
  int i, node_index, window_index;
  uint8_t rel_id, tuple_seq, query_id, seq_diff;
  uint8_t window_ofs, disable_join;
  WINDOW_ENTRY *p;
  QUERY_ENTRY *q;

DO_STACKTEST
  
  node_index = get_nodeindex(TUPLEGET_NODE(tuple));
  if (node_index < 0)
  {
    DBGERR("Cannot find node %d", TUPLEGET_NODE(tuple));
    pause_sim();
    return 1;
  }

  query_id = TUPLEGET_QUERY(tuple);
  q = &query_buffer[query_id];
  if (q->used == 0)
  {
    DBGERR("Query %d not used", query_id);
    return 1;
  }
  
  rel_id = TUPLEGET_REL(tuple);
  if (rel_id != REL_S && rel_id != REL_T)
  {
    DBGERR("No support for nested joins");
    pause_sim();
    return 1;
  }

  tuple_seq = TUPLEGET_SEQ(tuple);
  window_index = get_windowindex(node_index, query_id, rel_id);
  if (window_index < 0)
  {
    DBGERR("Cannot find window for node %d, query %d, rel %d", TUPLEGET_NODE(tuple), query_id, rel_id);
    pause_sim();
    return 1;
  }

  p = &windowjoin_windows[window_index];
  if (tuple_len != p->width)
  {
    DBGERR("Invalid tuple width: %d (expected %d)", tuple_len, p->width);
    pause_sim();
    return 1;
  }

  disable_join = TUPLEGET_DISABLEJOIN(tuple);
  seq_diff = ((0x7f+(uint16_t) tuple_seq-p->max_seq) % 0x7f);
  if (seq_diff > 0x3f && seq_diff <= 0x7f-p->height)
  {
    DBGOUT("Tuple too old (%d, %d). Discarding", tuple_seq, p->max_seq);
    return 0;
  }

  if (p->window == NULL)
  {
    p->window = (ATTRIBUTE *) calloc(sizeof(ATTRIBUTE), p->width*p->height);
    memset(p->window, 0xff, sizeof(ATTRIBUTE)*p->width*p->height);
  }

  if (seq_diff <= 0x3f)
  {
    if (seq_diff >= p->height)
      memset(p->window, 0xff, sizeof(ATTRIBUTE)*p->width*p->height);
    else {
      for (i = 0; i < p->height-seq_diff; i++)
        memcpy(&p->window[i*p->width], &p->window[(i+seq_diff)*p->width], sizeof(ATTRIBUTE)*p->width);
      memset(&p->window[(p->height-seq_diff)*p->width], 0xff, sizeof(ATTRIBUTE)*seq_diff*p->width);
    }
    p->max_seq = tuple_seq;
  }

  window_ofs = p->height-p->max_seq+tuple_seq-1;
  memcpy(&p->window[window_ofs*p->width], tuple, sizeof(ATTRIBUTE)*p->width);

  if (disable_join == 0)
  {
    do_join(q, window_index, window_ofs);

    for (i = 0; i < MAX_JOINPAIRS; i++)
      if (windowjoin_pairs[i].used && windowjoin_pairs[i].window_index[rel_id] == window_index &&
        windowjoin_pairs[i].tuple_count[rel_id] < MAX_ESTTUPLECOUNT)
        windowjoin_pairs[i].tuple_count[rel_id]++;
  }

  display_joinbuffer();

  return 0;
}

void estimate_sel(QUERY_ENTRY *q)
{
  int i, j, k, target_nodeindex;
  ADDRESS source_node, target_node;
  JOINPAIRS_ENTRY *p;
  uint16_t sel_est[REL_COUNT];
  ESTUPDATE_MSGHI msg_out;
  uint8_t buffer[sizeof(msg_out)];
  int size;
  ROUTING_PREDICATE predicate;
  uint32_t t, t1;

DO_STACKTEST

  for (i = 0; i < MAX_JOINPAIRS; i++)
  {
    p = &windowjoin_pairs[i];
    if (p->used && windowjoin_windows[p->window_index[REL_S]].query_id == q->query_id)
    {
      t1 = timer_cycles - p->start_time;
      t = t1/q->spec.params[PARAM_SAMPLEINTERVAL];

      if (t > 0 && (t % EST_WINDOW) == 0)
      {
        sel_est[REL_J] = (p->tuple_count[REL_J] == 0) ? MAX_SEL : MAX(MIN_SEL, MIN(MAX_SEL,
          DIV_ROUND(q->spec.params[PARAM_WINDOWSIZE]*(p->tuple_count[REL_S]+p->tuple_count[REL_T]), p->tuple_count[REL_J])));
        for (j = 0; j < SRCREL_COUNT; j++)
          sel_est[j] = (p->tuple_count[j] == 0) ? MAX_SEL : MAX(MIN_SEL, MIN(MAX_SEL,
            DIV_ROUND(t1, q->spec.params[PARAM_SAMPLEINTERVAL]*p->tuple_count[j])));

        for (j = 0; j < REL_COUNT; j++)
          if (t == EST_WINDOW || p->sel_est[j] > (EST_SENS(j)+1)*sel_est[j]/EST_SENS(j) ||
            p->sel_est[j] < (EST_SENS(j)-1)*sel_est[j]/EST_SENS(j))
          {                  // make sure to send estimates at least once, or then they change sufficiently
            source_node = windowjoin_nodes[windowjoin_windows[p->window_index[REL_S]].node_index].node_id;
            target_nodeindex = windowjoin_windows[p->window_index[REL_T]].node_index;
            target_node = windowjoin_nodes[target_nodeindex].node_id;

            for (k = 0; k < REL_COUNT; k++)
              p->sel_est[k] = sel_est[k];
#ifdef TOSSIM
            DBGOUT("Estimated selectivities updated: (%d, %d) = (%d, %d, %d)", source_node, target_node,
              sel_est[REL_S], sel_est[REL_T], sel_est[REL_J]);
            DBGOUT("Tuple counts: (%d, %d, %d), %d, %d", p->tuple_count[REL_S], p->tuple_count[REL_T], p->tuple_count[REL_J],
              timer_cycles, p->start_time);
            join_logselest(q->query_id, source_node, target_node, sel_est[REL_S], sel_est[REL_T], sel_est[REL_J]);
#endif

            msg_out.cmd = CMDHI_ESTUPDATE;
            msg_out.query_id = q->query_id;
            msg_out.source_node = source_node;
            for (k = 0; k < REL_COUNT; k++)
              msg_out.sel_est[k] = sel_est[k];

            DBGOUT("Sending sel update message to %d", target_node);
            size = marshal_estupdatemsghi(&msg_out, buffer);
            if (windowjoin_nodes[target_nodeindex].path_len == 0 && target_node != TOS_NODE_ID)
            {
              DBGOUT("Message sent using base flood");
              predicate.summary_index = find_idsummary();
              predicate.value = target_node;
              predicate.value_ext = 0;
              init_baseflood(1, &predicate, size, buffer);
            } else init_forward(TOS_NODE_ID, TOS_NODE_ID, target_node,
              windowjoin_nodes[target_nodeindex].path_len, windowjoin_nodes[target_nodeindex].path, FORWARD_DESTRUCTIVE,
                size, buffer, 0, NULL, ROUTE_SUCCESS);

            break;
          }

        if ((t1 % EST_RENORMALIZE) == 0)
        {
          p->start_time = (timer_cycles + p->start_time)/2;
          for (j = 0; j < REL_COUNT; j++)
            p->tuple_count[j] /= 2;

          DBGOUT("Renormalized tuple counts to (%d, %d, %d), start time to %d", p->tuple_count[REL_S],
            p->tuple_count[REL_T], p->tuple_count[REL_J], p->start_time);
        }
      }
    }  
  }
}

void update_windowjoin(QUERY_ENTRY *q)
{
  int j, k;

DO_STACKTEST

  if (q->options & QUERYOPT_ESTSEL && (timer_cycles % q->spec.params[PARAM_SAMPLEINTERVAL]) == 0)
    estimate_sel(q);

  if (q->spec.params[PARAM_WINDOWADVANCETYPE] == WINDOWADVANCE_TIME &&
    timer_cycles % q->spec.params[PARAM_WINDOWADVANCETIME] == 0)
    for (j = 0; j < MAX_JOINWINDOWS; j++)
      if (windowjoin_windows[j].used && windowjoin_windows[j].query_id == q->query_id)
        for (k = 0; k < windowjoin_windows[j].height; k++)
          if (windowjoin_windows[j].window[k*windowjoin_windows[j].width] != 0xffff)
          {
            DBGOUT("Erasing entry %d in window %d for query %d", k, j, q->query_id);
            windowjoin_windows[j].window[k*windowjoin_windows[j].width] = 0xffff;
            break;
          }
}

void stopquery_windowjoin(QUERY_ENTRY *p)
{
  int i, j;
  uint8_t node_used;

DO_STACKTEST
  
  for (i = 0; i < MAX_JOINWINDOWS; i++)
    if (windowjoin_windows[i].used && windowjoin_windows[i].query_id == p->query_id)
      windowjoin_windows[i].used = 0;

  for (i = 0; i < MAX_JOINPAIRS; i++)
    if (windowjoin_pairs[i].used)
      for (j = 0; j < SRCREL_COUNT; j++)
        if (windowjoin_windows[windowjoin_pairs[i].window_index[j]].used == 0)
        {
          windowjoin_pairs[i].used = 0;
          break;
        }

  for (i = 0; i < MAX_JOINNODES; i++)
    if (windowjoin_nodes[i].used)
    {
      node_used = 0;
      for (j = 0; j < MAX_JOINWINDOWS; j++)
        if (windowjoin_windows[j].used && windowjoin_windows[j].node_index == i)
        {
          node_used = 1;
          break;
        }

      windowjoin_nodes[i].used = node_used;
    }
}

uint8_t windowjoin_resolvepath(ADDRESS dest_node, ADDRESS *next_hop, ADDRESS *path, uint8_t *path_len)
{
  int i;

DO_STACKTEST

  for (i = 0; i < MAX_JOINNODES; i++)
    if (windowjoin_nodes[i].used && windowjoin_nodes[i].node_id == dest_node && windowjoin_nodes[i].path_len > 0)
    {
      *path_len = windowjoin_nodes[i].path_len;
      memcpy(path, windowjoin_nodes[i].path, *path_len*sizeof(ADDRESS));
      *next_hop = path[0];
      DBGOUT("Next hop retrieved from join buffer: %d", *next_hop);
      
      return 1;
    }

  return 0;
}


